package com.atguigu.cn.dataStream.api

import com.atguigu.cn.dataStream.api
import org.apache.flink.api.common.functions.{FilterFunction, MapFunction, RichFilterFunction, RichFlatMapFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @author yangshen
  *         transform 转换操作，流的拆分和合并
 * @date 2020/4/12 16:16
 */
object TransformTest {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    environment.setParallelism(1)

    //读入数据
    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")

    //transform 操作，并标记返回类型
    //1.基本转换算子和简单聚合算子
    val dataStream: DataStream[SensorReading] = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
    // 聚合操作
      val aggStream = dataStream.keyBy(0)
//        .sum(2)

    //或者用指定字段：  .keyBy("id").sum("temperature")

      //输出当前传感器最新的温度+10， 而时间戳是上一次数据时间+1
        .reduce( (x,y) => SensorReading(x.id, x.timestamp + 1, y.temperature + 10) )

//    aggStream.print()

    //2. 多流转换算子

    //2.1 split分流
    val splitStream = dataStream.split( data =>{
      if (data.temperature > 30) Seq("high") else Seq("low")
    })
    val highTempStream = splitStream.select("high")
    val lowTempStream = splitStream.select("low")
    val allTempStream = splitStream.select("high","low")
//    highTempStream.print("high")
//    lowTempStream.print("low")
//    allTempStream.print("all")

    //2.2 合并两条流
    val warningStream = highTempStream.map( data => (data.id, data.temperature) )
    //貌合神离，一国两制，数据类型可以不同(warning有两个字段，low有三个字段)
    //缺点：一次只能合并两个，合并后的类型是ConnectedStreams---转为DataStream---先转化为DataStream---才可以合并第三条流
    val connectedStreams = warningStream.connect(lowTempStream)

    //完全合并：且必须依赖 connectedStream
    val coMapDataStream = connectedStreams.map(
      warningData => (warningData._1, warningData._2, ""),
      lowData => (lowData.id, "healthy")
    )
    coMapDataStream.print("coMapDataStream data")

    //合并多个：但是类型必须一样
    val unionStream = highTempStream.union(lowTempStream)
//    unionStream.print()

    //3.函数类
    //3.1 匿名函数
    dataStream.filter(data => data.id.startsWith("sensor_1")).print()
    //和上一行意思一样，解释下面 _. 代表什么
    dataStream.filter(_.id.startsWith("sensor_1")).print()


    //3.2 自定义函数
//    dataStream.filter(new MyFilter()).print("MyFilter")
//    dataStream.map(new MyMap()).print("MyMap")
    dataStream.filter(new MyFilterUpGrade("sensor_1")).map(new MyMap()).print("MyMap")

    environment.execute("transform text")
  }
}
//自定义函数：MyFilter
class MyFilter extends FilterFunction[SensorReading] {
  override def filter(t: SensorReading): Boolean = {
    t.id.startsWith("sensor_1")
  }
}
//自定义函数：MyFilterUpGrade
class MyFilterUpGrade(inputKeyWord:String) extends FilterFunction[SensorReading] {
  override def filter(t: SensorReading): Boolean = {
    t.id.startsWith(inputKeyWord)
  }
}
//自定义函数：MyMap
class MyMap extends MapFunction[SensorReading,SensorReading]{
  override def map(t: SensorReading): SensorReading = {
    SensorReading(t.id, t.timestamp, t.temperature + 1)
  }
}
//富函数--包含生命周期
class MyRichMap extends RichMapFunction[SensorReading, String]{
  override def map(in: SensorReading): String = {
    "flink"
  }
}

//富函数--包含生命周期
class MyRichFlatMap extends RichFlatMapFunction[Int, (Int,Int)]{
  var subtaskIndex = 0
  //重写 RichMapFunction 的open方法
  override def open(configuration: Configuration): Unit = {
    //当前运行环境获取当前运行子任务的索引值
    subtaskIndex = getRuntimeContext.getIndexOfThisSubtask
    //以下可以做一下初始化工作，例如建立一个和HDFS，redis 的连接, 在创建的时候去连接，而不用等数据来了再去连接
  }

  override def flatMap(in: Int, out: Collector[(Int, Int)]): Unit = {
    if (in % 2 == subtaskIndex){
      //out.collect 表示输出
      out.collect((subtaskIndex,in))
    }
  }

  override def close(): Unit = {
    //以下做一些清理工作，例如断开和HDFS的连接
  }
}
